package com.niit.covid.reduceSide;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

import java.io.IOException;

/**
 * @author:Ys
 * @date: 2022年08月15日 17:43
 * @desc: 使用 mapper 处理订单数据和商品数据，输出的时候以 goodsId 商品编号作为 key。相
 * 同 goodsId 的商品和订单会到同一个 reduce 的同一个分组，在分组中进行订单和商品信息的关联合并。
 * 在 MapReduce 程序中可以通过 context 获取到当前处理的切片所属的文件名称。
 * 根据文件名来判断当前处理的是订单数据还是商品数据，以此来进行不同逻辑的输出。
 * join 处理完之后，最后可以再通过 MapReduce 程序排序功能，将属于同一笔订单的所有商品信息汇聚在一起。
 */
public class ReduceJoinMapper extends Mapper<LongWritable, Text, Text, Text> {

    Text outKey = new Text();//输出的Key是个Text  goodsId 商品id
    Text outValue = new Text();  // 商品信息 或 订单信息
    StringBuilder sb = new StringBuilder();//是String的一个升级版（工具类），在本项目用追加 apeend
    String filename = null; //文件名称
    //setup用于做初始化的，在map第一次执行前就仅会执行一次
    @Override
    protected void setup(Context context) throws IOException, InterruptedException {
        //获取当前处理的切片所属的文件名字
        FileSplit inputSplit = (FileSplit) context.getInputSplit();//获得输入的分片
        filename = inputSplit.getPath().getName();//根据输入的分片获得文件名称
        //在map方法中可以根据不同的文件名称对Value进行组成，带有特殊标识的Value
        System.out.println("当前正在处理的文件是：" + filename);
    }

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

        //设置字符串长度，此处用于清空数据
        sb.setLength(0);

        //切割处理输入数据
        String[] fields = value.toString().split("\\|");
        //判断处理的是哪个文件
        if (filename.contains("my_goods.txt")) {//处理的是商品数据
            // 100101|155083444927602|四川果冻橙6个约180g  （商品id、商品编号、商品名称）
            String goodsId = fields[0];
            String goodsNo = fields[1];
            String goodsName = fields[2];
            outKey.set(goodsId);
            StringBuilder append = sb.append(goodsNo).append("\t").append(goodsName);// 155083444927602 四川果冻橙6个约180g
            // goods#155083444927602 四川果冻橙6个约180g
            outValue.set(   sb.insert(0,"goods#").toString()          );

        } else {//处理的是订单数据
            //  2|113561|11192  （订单编号、商品id、实际支付价格）
            String goodsId = fields[1];
            outKey.set(goodsId);
            sb.append(fields[0]).append("\t").append(fields[2]);
            outValue.set( sb.insert(0,"order#").toString()    );
        }

        context.write(outKey,outValue);
    }

    //在所有map方法执行完成后，仅会执行一次，将最终数据放到cleanup进行输出
    @Override
    protected void cleanup(Context context) throws IOException, InterruptedException {
        super.cleanup(context);
    }
}

